[FLINK-39436][table] Allow late data in PTFs by fhueske · Pull Request #27935 · apache/flink

fhueske · 2026-04-15T17:01:26Z

What is the purpose of the change

Previously, late events (rowtime <= watermark) were silently dropped before reaching PTF eval(), and timer registrations for times <= watermark were also silently dropped. This change removes both restrictions.
These are breaking changes that are part of FLIP-565 which was approved.

I checked that the docs don't need to be adjusted because the earlier behavior of PTFs dropping late data was not documented. The new behavior is aligned with other functions in Flink.

Brief change log

ProcessTableRunner: remove the early-return guard in processEval() so that late events are passed to the PTF's eval() method.
WritableInternalTimeContext: remove the watermark check in registerOnTimeInternal() so that timers can be registered for past times. Such timers fire immediately at the next watermark advance, including when registered from within onTimer(). The previous guard also had an unintended side effect: any call to replaceNamedTimer() with a past time would delete the existing timer entry but then silently drop the new registration, leaving the named timer in a state where it appeared un-registered but the old timer was gone.
AbstractProcessTableOperator: remove the fired named timer's state entry before invoking onTimer() to prevent stale entries from accumulating in the named timers map state.

Verifying this change

Tests are updated to reflect the new semantics:

PROCESS_LATE_EVENTS: demonstrates that late events enter eval(), can register timers (including for past times), and that such timers fire immediately at the next watermark advance.
PROCESS_ROW_LATE_EVENTS (new): verifies the same for row-semantics PTFs.
PROCESS_OPTIONAL_ON_TIME / PROCESS_NAMED_TIMERS: updated to reflect that timer registrations for past times are no longer dropped. Previously, once the watermark passed the registered time, the timer was silently discarded; now it fires immediately.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): no
The public API, i.e., is any changed class annotated with @Public(Evolving): no
The serializers: no
The runtime per-record code paths (performance sensitive): no
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
The S3 file system connector: no

Documentation

Does this pull request introduce a new feature? no
If yes, how is the feature documented? n/a

flinkbot · 2026-04-15T17:05:31Z

CI report:

521b017 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

gustavodemorais

The PR looks good in general to me!

Two nits: can you rename the PR title and commit title to use this format [FLINK-39436][table]?

gustavodemorais · 2026-04-16T10:40:06Z

                                            "+I[Alice, {Processing input row +I[Alice, 1, 1970-01-01T00:00:00.001Z] at time 1 watermark null}, 1970-01-01T00:00:00.001Z]",
                                            "+I[Bob, {Processing input row +I[Bob, 2, 1970-01-01T00:00:00.002Z] at time 2 watermark null}, 1970-01-01T00:00:00.002Z]")
                                    .consumedAfterRestore(
                                            "+I[Bob, {Timer timeout1 fired at time 1 watermark 9223372036854775807}, 1970-01-01T00:00:00.001Z]",


One thing that was a bit confusing for me is why we have watermark 9223372036854775807 here? Do we emit this max watermark only for testing or do our sources actually emit this when all values have been read? I wondered if that basically makes the user not know if something is late anymore since

Sources emit this when all values have been read. It will flush all remaining timers and windows and mark the job as complete.

Makes sense. This is an unrelated topic to the PR so definitely not a blocker: but I guess it makes sense for users to be aware of this so they can write their logic accordingly as well? Maybe it's already documented somewhere and then there's nothing to do 🙂

And for this PR: maybe add a short comment Sources emit 9223372036854775807 as a MAX watermark after all values have been read. It makes understanding the test a bit easier

AFAICT this is a property of the testing framework.
I can see that this looks surprising, but I'm not sure if it makes sense to put a comment everywhere the framework is being used.

gustavodemorais · 2026-04-16T11:41:56Z

One thing worth noting: removing the watermark guard means past-time timers registered from onTimer() fire immediately (within the same advanceWatermark loop). But it also means something like this lead to infinite loops:

 public void onTimer(OnTimerContext ctx) {
      // "remind me again in the past" = fires immediately = infinite loop
      ctx.timeContext(Long.class).registerOnTime(0L);
  }

AI says this is the same behavior as standard Flink KeyedProcessFunction, so the design is consistent - just wanted to flag that PTF users were previously protected from this and they may shoot themselves in the foot more easily now. Maybe worth a short note in the timers section of ptfs.md along the lines of "timers registered at or below the current watermark fire immediately at the next watermark advance; avoid unconditional re-registration of past-time timers from onTimer() as this will cause an infinite loop."

twalthr

Thanks for the PR @fhueske. I added some minor comments. Could we update the JavaDoc in ProcessTableFunction and the documentation for late events. Late events deserve a dedicated section, before * <h2>Efficiency and Design Principles</h2>.

twalthr · 2026-04-16T11:07:51Z

+            // Register at wm+1 to always target the immediate next watermark: the timer fires
+            // exactly once per watermark advance, and each new row re-registers the timer for the
+            // following watermark step, demonstrating repeated timer re-registration.
+            long timer = wm == null || wm < 0 ? 1 : wm + 1;


wouldn't this be more correct and allowing for watermarks before epoch.

Suggested change

long timer = wm == null || wm < 0 ? 1 : wm + 1;

long timer = wm == null ? Long.MIN_VALUE + 1: wm + 1;

This is just a function for testing purposes and we assert against the behavior that we implement.
So I think it doesn't really matter.
Or is there an (edge) case that would be covered with your suggestion that isn't with the current version?

fhueske · 2026-04-16T17:19:26Z

Thanks for your reviews @gustavodemorais and @twalthr.
I've addressed your comments and updated the PR.

gustavodemorais

The PR looks good to me! Thanks, @fhueske

I think you have to run spotless to make the CI green

twalthr

Thanks for the update @fhueske, I had two last comments. Should be good in the next iteration.

Au-Miner · 2026-04-20T07:36:15Z

+Registering a timer for a time that is less than or equal to the current watermark is allowed.
+If registered from within `eval()`, the timer fires on the next watermark advance. If registered
+from within `onTimer()`, the timer fires immediately after the current timer finishes. Note that
+unconditionally re-registering a past-time timer from within `onTimer()` causes an infinite loop.


Explicitly warning in Javadoc that 'unconditionally re-registering a past-time timer from within onTimer() causes an infinite loop';
Is it necessary to consider adding protection, such as adding detection logic.
Alternatively, it may be necessary to add this infinite loop error test in the test, such as adding the timer for 'alice-again' in LateTimersFunction # eval

This is the same behavior as for other functions with access to timers such as KeyedProcessFunction.
Before this change, we ignored the registrations of timers with a timestamp < current_watermark, which is inconsistent with Flink's other functions and surprising for Flink users.

We might be able to introduce guards against such behavior, but I don't think it is necessary. I don't think this is a common problem and adding logic to detect and prevent such logic would cause more problems than it solves.

Au-Miner · 2026-04-20T07:45:01Z

@@ -1115,35 +1116,61 @@ public class ProcessTableFunctionTestPrograms {
    public static final TableTestProgram PROCESS_LATE_EVENTS =


Is it possible to add PROCESS_LATE_EVENTS_RESTORE.

fhueske · 2026-04-20T14:11:09Z

Thanks for the review @Au-Miner.
I've added a restore test with late records.

Best, Fabian

twalthr

Thanks for the update @fhueske. The PR looks good to me.

Previously, late events (rowtime <= watermark) were silently dropped before reaching PTF eval(), and timer registrations for times <= watermark were also silently dropped. This change removes both restrictions: - ProcessTableRunner: remove the early-return guard in processEval() so that late events are passed to the PTF's eval() method. - WritableInternalTimeContext: remove the watermark check in registerOnTimeInternal() so that timers can be registered for past times. Such timers fire immediately at the next watermark advance, including when registered from within onTimer(). The previous guard also had an unintended side effect: any call to replaceNamedTimer() with a past time would delete the existing timer entry but then silently drop the new registration, leaving the named timer in a state where it appeared un-registered but the old timer was gone. - AbstractProcessTableOperator: remove the fired named timer's state entry before invoking onTimer() to prevent stale entries from accumulating in the named timers map state. Tests are updated to reflect the new semantics: - PROCESS_LATE_EVENTS: demonstrates that late events enter eval(), can register timers (including for past times), and that such timers fire immediately at the next watermark advance. - PROCESS_ROW_LATE_EVENTS (new): verifies the same for row-semantics PTFs. - PROCESS_OPTIONAL_ON_TIME / PROCESS_NAMED_TIMERS: updated to reflect that timer registrations for past times are no longer dropped. Previously, once the watermark passed the registered time, the timer was silently discarded; now it fires immediately.

fhueske requested a review from twalthr April 15, 2026 17:12

fhueske marked this pull request as ready for review April 15, 2026 17:13

gustavodemorais reviewed Apr 16, 2026

View reviewed changes

Comment thread .../org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java Outdated

twalthr reviewed Apr 16, 2026

View reviewed changes

github-actions Bot added the community-reviewed PR has been reviewed by the community. label Apr 16, 2026

fhueske force-pushed the fhueske-FLINK-39436-Allow-late-data-in-PTFs branch from 857e77c to a08f393 Compare April 16, 2026 17:18

fhueske changed the title ~~[FLINK-39436] [sql] Allow late data in PTFs~~ [FLINK-39436][table] Allow late data in PTFs Apr 16, 2026

fhueske requested review from gustavodemorais and twalthr April 16, 2026 17:18

gustavodemorais approved these changes Apr 16, 2026

View reviewed changes

fhueske force-pushed the fhueske-FLINK-39436-Allow-late-data-in-PTFs branch from a08f393 to 57b157d Compare April 17, 2026 06:38

twalthr reviewed Apr 17, 2026

View reviewed changes

Comment thread docs/content/docs/dev/table/functions/ptfs.md

Comment thread docs/content/docs/dev/table/functions/ptfs.md Outdated

fhueske force-pushed the fhueske-FLINK-39436-Allow-late-data-in-PTFs branch from 57b157d to 1781123 Compare April 17, 2026 17:55

Au-Miner reviewed Apr 20, 2026

View reviewed changes

fhueske force-pushed the fhueske-FLINK-39436-Allow-late-data-in-PTFs branch from 1781123 to e63b9b3 Compare April 20, 2026 14:10

twalthr approved these changes Apr 21, 2026

View reviewed changes

fhueske added 4 commits April 21, 2026 14:26

[FLINK-39436][table] Address reviewer comments

7410d18

[FLINK-39436][table] Address comments #2

dd8d9ff

[FLINK-39436][table] Add PTF restore test with late data

521b017

fhueske force-pushed the fhueske-FLINK-39436-Allow-late-data-in-PTFs branch from e63b9b3 to 521b017 Compare April 21, 2026 12:27

fhueske merged commit 9373256 into master Apr 21, 2026
20 of 22 checks passed

fhueske deleted the fhueske-FLINK-39436-Allow-late-data-in-PTFs branch April 21, 2026 19:28

fhueske mentioned this pull request Apr 22, 2026

[BP-2.3][FLINK-39436][FLINK-39437][table] Backport of two PTF fixes #27998

Merged

	long timer = wm == null \|\| wm < 0 ? 1 : wm + 1;
	long timer = wm == null ? Long.MIN_VALUE + 1: wm + 1;

		@@ -1115,35 +1116,61 @@ public class ProcessTableFunctionTestPrograms {
		public static final TableTestProgram PROCESS_LATE_EVENTS =

Conversation

fhueske commented Apr 15, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Uh oh!

flinkbot commented Apr 15, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

gustavodemorais left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

gustavodemorais commented Apr 16, 2026

Uh oh!

twalthr left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

fhueske commented Apr 16, 2026

Uh oh!

gustavodemorais left a comment

Choose a reason for hiding this comment

Uh oh!

twalthr left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

fhueske commented Apr 20, 2026

Uh oh!

twalthr left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

fhueske commented Apr 15, 2026 •

edited

Loading

flinkbot commented Apr 15, 2026 •

edited

Loading